#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
FastAPI数据库操作示例 - 异步数据库访问详解

本文件展示FastAPI与各种数据库的异步操作，包括SQLAlchemy异步、
MongoDB异步、Redis异步等，并与SpringBoot的数据访问进行对比。

FastAPI数据库特点：
1. 支持多种异步数据库驱动
2. SQLAlchemy 2.0异步支持
3. Pydantic模型验证
4. 依赖注入数据库会话
5. 自动API文档生成

作者: Assistant
日期: 2024
"""

import asyncio
import logging
from datetime import datetime, timedelta
from typing import List, Optional, Dict, Any
from contextlib import asynccontextmanager

# FastAPI相关导入
from fastapi import FastAPI, Depends, HTTPException, Query, Path, BackgroundTasks
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field, validator
from pydantic.types import EmailStr

# SQLAlchemy异步支持
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from sqlalchemy import Column, Integer, String, DateTime, Boolean, Text, ForeignKey, select, update, delete
from sqlalchemy.sql import func

# MongoDB异步支持
import motor.motor_asyncio
from pymongo import IndexModel
from bson import ObjectId

# Redis异步支持
import aioredis
from aioredis import Redis

# 其他工具
import json
import uuid
from functools import lru_cache

# ============================================================================
# 1. 数据库配置和连接
# ============================================================================

# 数据库配置
class DatabaseConfig:
    """数据库配置类"""
    
    # PostgreSQL异步连接
    POSTGRES_URL = "postgresql+asyncpg://user:password@localhost:5432/fastapi_db"
    
    # MySQL异步连接
    MYSQL_URL = "mysql+aiomysql://user:password@localhost:3306/fastapi_db"
    
    # SQLite异步连接（用于演示）
    SQLITE_URL = "sqlite+aiosqlite:///./fastapi_demo.db"
    
    # MongoDB连接
    MONGODB_URL = "mongodb://localhost:27017"
    MONGODB_DB = "fastapi_db"
    
    # Redis连接
    REDIS_URL = "redis://localhost:6379/0"

# SQLAlchemy异步引擎
engine = create_async_engine(
    DatabaseConfig.SQLITE_URL,
    echo=True,  # 开发环境显示SQL
    future=True
)

# 异步会话工厂
AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False
)

# MongoDB异步客户端
mongo_client = motor.motor_asyncio.AsyncIOMotorClient(DatabaseConfig.MONGODB_URL)
mongo_db = mongo_client[DatabaseConfig.MONGODB_DB]

# Redis异步客户端
redis_client: Optional[Redis] = None

# ============================================================================
# 2. SQLAlchemy模型定义
# ============================================================================

class Base(DeclarativeBase):
    """SQLAlchemy基础模型"""
    pass

class User(Base):
    """用户模型"""
    __tablename__ = "users"
    
    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    email: Mapped[str] = mapped_column(String(100), unique=True, index=True)
    full_name: Mapped[Optional[str]] = mapped_column(String(100))
    hashed_password: Mapped[str] = mapped_column(String(100))
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=func.now())
    updated_at: Mapped[datetime] = mapped_column(DateTime, default=func.now(), onupdate=func.now())
    
    # 关系
    posts: Mapped[List["Post"]] = relationship("Post", back_populates="author")

class Post(Base):
    """文章模型"""
    __tablename__ = "posts"
    
    id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
    title: Mapped[str] = mapped_column(String(200))
    content: Mapped[str] = mapped_column(Text)
    author_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"))
    is_published: Mapped[bool] = mapped_column(Boolean, default=False)
    created_at: Mapped[datetime] = mapped_column(DateTime, default=func.now())
    updated_at: Mapped[datetime] = mapped_column(DateTime, default=func.now(), onupdate=func.now())
    
    # 关系
    author: Mapped["User"] = relationship("User", back_populates="posts")

# ============================================================================
# 3. Pydantic模型定义
# ============================================================================

class UserBase(BaseModel):
    """用户基础模型"""
    username: str = Field(..., min_length=3, max_length=50, description="用户名")
    email: EmailStr = Field(..., description="邮箱地址")
    full_name: Optional[str] = Field(None, max_length=100, description="全名")
    
    @validator('username')
    def validate_username(cls, v):
        if not v.isalnum():
            raise ValueError('用户名只能包含字母和数字')
        return v

class UserCreate(UserBase):
    """创建用户模型"""
    password: str = Field(..., min_length=6, description="密码")

class UserUpdate(BaseModel):
    """更新用户模型"""
    email: Optional[EmailStr] = None
    full_name: Optional[str] = None
    is_active: Optional[bool] = None

class UserResponse(UserBase):
    """用户响应模型"""
    id: int
    is_active: bool
    created_at: datetime
    updated_at: datetime
    
    class Config:
        from_attributes = True

class PostBase(BaseModel):
    """文章基础模型"""
    title: str = Field(..., min_length=1, max_length=200, description="文章标题")
    content: str = Field(..., min_length=1, description="文章内容")
    is_published: bool = Field(False, description="是否发布")

class PostCreate(PostBase):
    """创建文章模型"""
    pass

class PostUpdate(BaseModel):
    """更新文章模型"""
    title: Optional[str] = None
    content: Optional[str] = None
    is_published: Optional[bool] = None

class PostResponse(PostBase):
    """文章响应模型"""
    id: int
    author_id: int
    created_at: datetime
    updated_at: datetime
    author: UserResponse
    
    class Config:
        from_attributes = True

class ApiResponse(BaseModel):
    """API响应格式"""
    success: bool
    message: str
    data: Optional[Any] = None
    timestamp: datetime = Field(default_factory=datetime.now)

# MongoDB文档模型
class MongoUser(BaseModel):
    """MongoDB用户文档"""
    id: Optional[str] = Field(None, alias="_id")
    username: str
    email: str
    full_name: Optional[str] = None
    created_at: datetime = Field(default_factory=datetime.now)
    metadata: Dict[str, Any] = Field(default_factory=dict)
    
    class Config:
        populate_by_name = True
        json_encoders = {
            ObjectId: str
        }

# ============================================================================
# 4. 数据库依赖注入
# ============================================================================

async def get_db() -> AsyncSession:
    """获取数据库会话"""
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

async def get_mongo_db():
    """获取MongoDB数据库"""
    return mongo_db

async def get_redis() -> Redis:
    """获取Redis客户端"""
    global redis_client
    if redis_client is None:
        redis_client = await aioredis.from_url(DatabaseConfig.REDIS_URL)
    return redis_client

# ============================================================================
# 5. 数据访问层 (Repository Pattern)
# ============================================================================

class UserRepository:
    """用户数据访问层"""
    
    def __init__(self, db: AsyncSession):
        self.db = db
    
    async def create_user(self, user_data: UserCreate) -> User:
        """创建用户"""
        # 简单的密码哈希（实际应用中应使用bcrypt等）
        hashed_password = f"hashed_{user_data.password}"
        
        db_user = User(
            username=user_data.username,
            email=user_data.email,
            full_name=user_data.full_name,
            hashed_password=hashed_password
        )
        
        self.db.add(db_user)
        await self.db.commit()
        await self.db.refresh(db_user)
        return db_user
    
    async def get_user_by_id(self, user_id: int) -> Optional[User]:
        """根据ID获取用户"""
        result = await self.db.execute(
            select(User).where(User.id == user_id)
        )
        return result.scalar_one_or_none()
    
    async def get_user_by_username(self, username: str) -> Optional[User]:
        """根据用户名获取用户"""
        result = await self.db.execute(
            select(User).where(User.username == username)
        )
        return result.scalar_one_or_none()
    
    async def get_users(self, skip: int = 0, limit: int = 100) -> List[User]:
        """获取用户列表"""
        result = await self.db.execute(
            select(User).offset(skip).limit(limit)
        )
        return result.scalars().all()
    
    async def update_user(self, user_id: int, user_data: UserUpdate) -> Optional[User]:
        """更新用户"""
        # 构建更新数据
        update_data = {k: v for k, v in user_data.dict().items() if v is not None}
        if not update_data:
            return await self.get_user_by_id(user_id)
        
        update_data['updated_at'] = datetime.now()
        
        await self.db.execute(
            update(User).where(User.id == user_id).values(**update_data)
        )
        await self.db.commit()
        
        return await self.get_user_by_id(user_id)
    
    async def delete_user(self, user_id: int) -> bool:
        """删除用户"""
        result = await self.db.execute(
            delete(User).where(User.id == user_id)
        )
        await self.db.commit()
        return result.rowcount > 0
    
    async def search_users(self, query: str) -> List[User]:
        """搜索用户"""
        result = await self.db.execute(
            select(User).where(
                User.username.contains(query) | 
                User.full_name.contains(query) |
                User.email.contains(query)
            )
        )
        return result.scalars().all()

class PostRepository:
    """文章数据访问层"""
    
    def __init__(self, db: AsyncSession):
        self.db = db
    
    async def create_post(self, post_data: PostCreate, author_id: int) -> Post:
        """创建文章"""
        db_post = Post(
            title=post_data.title,
            content=post_data.content,
            is_published=post_data.is_published,
            author_id=author_id
        )
        
        self.db.add(db_post)
        await self.db.commit()
        await self.db.refresh(db_post)
        return db_post
    
    async def get_post_by_id(self, post_id: int) -> Optional[Post]:
        """根据ID获取文章（包含作者信息）"""
        result = await self.db.execute(
            select(Post).options(
                # 预加载作者信息
                # selectinload(Post.author)
            ).where(Post.id == post_id)
        )
        return result.scalar_one_or_none()
    
    async def get_posts(self, skip: int = 0, limit: int = 100, published_only: bool = False) -> List[Post]:
        """获取文章列表"""
        query = select(Post)
        
        if published_only:
            query = query.where(Post.is_published == True)
        
        query = query.offset(skip).limit(limit).order_by(Post.created_at.desc())
        
        result = await self.db.execute(query)
        return result.scalars().all()
    
    async def get_posts_by_author(self, author_id: int) -> List[Post]:
        """获取指定作者的文章"""
        result = await self.db.execute(
            select(Post).where(Post.author_id == author_id)
        )
        return result.scalars().all()
    
    async def update_post(self, post_id: int, post_data: PostUpdate) -> Optional[Post]:
        """更新文章"""
        update_data = {k: v for k, v in post_data.dict().items() if v is not None}
        if not update_data:
            return await self.get_post_by_id(post_id)
        
        update_data['updated_at'] = datetime.now()
        
        await self.db.execute(
            update(Post).where(Post.id == post_id).values(**update_data)
        )
        await self.db.commit()
        
        return await self.get_post_by_id(post_id)
    
    async def delete_post(self, post_id: int) -> bool:
        """删除文章"""
        result = await self.db.execute(
            delete(Post).where(Post.id == post_id)
        )
        await self.db.commit()
        return result.rowcount > 0

# ============================================================================
# 6. MongoDB操作服务
# ============================================================================

class MongoUserService:
    """MongoDB用户服务"""
    
    def __init__(self, db):
        self.collection = db.users
    
    async def create_user(self, user_data: dict) -> str:
        """创建用户"""
        user_data['created_at'] = datetime.now()
        result = await self.collection.insert_one(user_data)
        return str(result.inserted_id)
    
    async def get_user_by_id(self, user_id: str) -> Optional[dict]:
        """根据ID获取用户"""
        try:
            user = await self.collection.find_one({"_id": ObjectId(user_id)})
            if user:
                user['_id'] = str(user['_id'])
            return user
        except Exception:
            return None
    
    async def get_users(self, skip: int = 0, limit: int = 100) -> List[dict]:
        """获取用户列表"""
        cursor = self.collection.find().skip(skip).limit(limit)
        users = await cursor.to_list(length=limit)
        
        for user in users:
            user['_id'] = str(user['_id'])
        
        return users
    
    async def update_user(self, user_id: str, update_data: dict) -> bool:
        """更新用户"""
        try:
            result = await self.collection.update_one(
                {"_id": ObjectId(user_id)},
                {"$set": update_data}
            )
            return result.modified_count > 0
        except Exception:
            return False
    
    async def delete_user(self, user_id: str) -> bool:
        """删除用户"""
        try:
            result = await self.collection.delete_one({"_id": ObjectId(user_id)})
            return result.deleted_count > 0
        except Exception:
            return False
    
    async def search_users(self, query: str) -> List[dict]:
        """搜索用户"""
        cursor = self.collection.find({
            "$or": [
                {"username": {"$regex": query, "$options": "i"}},
                {"email": {"$regex": query, "$options": "i"}},
                {"full_name": {"$regex": query, "$options": "i"}}
            ]
        })
        
        users = await cursor.to_list(length=100)
        for user in users:
            user['_id'] = str(user['_id'])
        
        return users

# ============================================================================
# 7. Redis缓存服务
# ============================================================================

class CacheService:
    """Redis缓存服务"""
    
    def __init__(self, redis: Redis):
        self.redis = redis
    
    async def get(self, key: str) -> Optional[str]:
        """获取缓存"""
        try:
            value = await self.redis.get(key)
            return value.decode('utf-8') if value else None
        except Exception as e:
            logging.error(f"Redis get error: {e}")
            return None
    
    async def set(self, key: str, value: str, expire: int = 3600) -> bool:
        """设置缓存"""
        try:
            await self.redis.set(key, value, ex=expire)
            return True
        except Exception as e:
            logging.error(f"Redis set error: {e}")
            return False
    
    async def delete(self, key: str) -> bool:
        """删除缓存"""
        try:
            result = await self.redis.delete(key)
            return result > 0
        except Exception as e:
            logging.error(f"Redis delete error: {e}")
            return False
    
    async def get_json(self, key: str) -> Optional[dict]:
        """获取JSON缓存"""
        value = await self.get(key)
        if value:
            try:
                return json.loads(value)
            except json.JSONDecodeError:
                return None
        return None
    
    async def set_json(self, key: str, value: dict, expire: int = 3600) -> bool:
        """设置JSON缓存"""
        try:
            json_str = json.dumps(value, default=str)
            return await self.set(key, json_str, expire)
        except Exception as e:
            logging.error(f"Redis set_json error: {e}")
            return False
    
    async def increment(self, key: str, amount: int = 1) -> int:
        """递增计数器"""
        try:
            return await self.redis.incrby(key, amount)
        except Exception as e:
            logging.error(f"Redis increment error: {e}")
            return 0
    
    async def get_or_set(self, key: str, factory_func, expire: int = 3600) -> Any:
        """获取或设置缓存"""
        # 先尝试从缓存获取
        cached_value = await self.get_json(key)
        if cached_value is not None:
            return cached_value
        
        # 缓存不存在，调用工厂函数生成数据
        if asyncio.iscoroutinefunction(factory_func):
            value = await factory_func()
        else:
            value = factory_func()
        
        # 设置缓存
        if value is not None:
            await self.set_json(key, value, expire)
        
        return value

# ============================================================================
# 8. 应用程序生命周期管理
# ============================================================================

@asynccontextmanager
async def lifespan(app: FastAPI):
    """应用程序生命周期管理"""
    # 启动时执行
    print("🚀 启动FastAPI数据库示例应用...")
    
    # 创建数据库表
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    
    # 初始化Redis连接
    global redis_client
    try:
        redis_client = await aioredis.from_url(DatabaseConfig.REDIS_URL)
        await redis_client.ping()
        print("✅ Redis连接成功")
    except Exception as e:
        print(f"❌ Redis连接失败: {e}")
        redis_client = None
    
    # 创建MongoDB索引
    try:
        await mongo_db.users.create_indexes([
            IndexModel("username", unique=True),
            IndexModel("email", unique=True),
            IndexModel("created_at")
        ])
        print("✅ MongoDB索引创建成功")
    except Exception as e:
        print(f"❌ MongoDB索引创建失败: {e}")
    
    yield
    
    # 关闭时执行
    print("🛑 关闭FastAPI数据库示例应用...")
    
    # 关闭Redis连接
    if redis_client:
        await redis_client.close()
    
    # 关闭MongoDB连接
    mongo_client.close()
    
    # 关闭数据库引擎
    await engine.dispose()

# ============================================================================
# 9. FastAPI应用程序创建
# ============================================================================

app = FastAPI(
    title="FastAPI数据库操作示例",
    description="展示FastAPI与各种数据库的异步操作",
    version="1.0.0",
    lifespan=lifespan
)

# 添加CORS中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ============================================================================
# 10. SQLAlchemy路由
# ============================================================================

@app.get("/", response_model=ApiResponse)
async def root():
    """首页"""
    return ApiResponse(
        success=True,
        message="欢迎使用FastAPI数据库操作示例",
        data={
            "features": [
                "SQLAlchemy异步操作",
                "MongoDB异步操作",
                "Redis缓存",
                "依赖注入",
                "数据验证",
                "自动API文档"
            ],
            "endpoints": {
                "sql": "/sql/users",
                "mongo": "/mongo/users",
                "cache": "/cache/demo",
                "docs": "/docs"
            }
        }
    )

# SQLAlchemy用户路由
@app.post("/sql/users", response_model=ApiResponse)
async def create_sql_user(
    user_data: UserCreate,
    db: AsyncSession = Depends(get_db)
):
    """创建用户（SQLAlchemy）"""
    repo = UserRepository(db)
    
    # 检查用户名是否已存在
    existing_user = await repo.get_user_by_username(user_data.username)
    if existing_user:
        raise HTTPException(status_code=400, detail="用户名已存在")
    
    user = await repo.create_user(user_data)
    
    return ApiResponse(
        success=True,
        message="用户创建成功",
        data=UserResponse.from_orm(user).dict()
    )

@app.get("/sql/users/{user_id}", response_model=ApiResponse)
async def get_sql_user(
    user_id: int = Path(..., description="用户ID"),
    db: AsyncSession = Depends(get_db),
    redis: Redis = Depends(get_redis)
):
    """获取用户（SQLAlchemy + Redis缓存）"""
    cache_service = CacheService(redis)
    cache_key = f"user:{user_id}"
    
    # 先从缓存获取
    cached_user = await cache_service.get_json(cache_key)
    if cached_user:
        return ApiResponse(
            success=True,
            message="从缓存获取用户成功",
            data={**cached_user, "from_cache": True}
        )
    
    # 缓存不存在，从数据库获取
    repo = UserRepository(db)
    user = await repo.get_user_by_id(user_id)
    
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    
    user_data = UserResponse.from_orm(user).dict()
    
    # 设置缓存
    await cache_service.set_json(cache_key, user_data, expire=300)
    
    return ApiResponse(
        success=True,
        message="获取用户成功",
        data={**user_data, "from_cache": False}
    )

@app.get("/sql/users", response_model=ApiResponse)
async def list_sql_users(
    skip: int = Query(0, ge=0, description="跳过记录数"),
    limit: int = Query(10, ge=1, le=100, description="限制记录数"),
    search: Optional[str] = Query(None, description="搜索关键词"),
    db: AsyncSession = Depends(get_db)
):
    """获取用户列表（SQLAlchemy）"""
    repo = UserRepository(db)
    
    if search:
        users = await repo.search_users(search)
    else:
        users = await repo.get_users(skip, limit)
    
    users_data = [UserResponse.from_orm(user).dict() for user in users]
    
    return ApiResponse(
        success=True,
        message=f"获取用户列表成功，共 {len(users_data)} 条记录",
        data={
            "users": users_data,
            "pagination": {
                "skip": skip,
                "limit": limit,
                "total": len(users_data)
            }
        }
    )

@app.put("/sql/users/{user_id}", response_model=ApiResponse)
async def update_sql_user(
    user_id: int,
    user_data: UserUpdate,
    db: AsyncSession = Depends(get_db),
    redis: Redis = Depends(get_redis)
):
    """更新用户（SQLAlchemy）"""
    repo = UserRepository(db)
    user = await repo.update_user(user_id, user_data)
    
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    
    # 清除缓存
    cache_service = CacheService(redis)
    await cache_service.delete(f"user:{user_id}")
    
    return ApiResponse(
        success=True,
        message="用户更新成功",
        data=UserResponse.from_orm(user).dict()
    )

@app.delete("/sql/users/{user_id}", response_model=ApiResponse)
async def delete_sql_user(
    user_id: int,
    db: AsyncSession = Depends(get_db),
    redis: Redis = Depends(get_redis)
):
    """删除用户（SQLAlchemy）"""
    repo = UserRepository(db)
    success = await repo.delete_user(user_id)
    
    if not success:
        raise HTTPException(status_code=404, detail="用户不存在")
    
    # 清除缓存
    cache_service = CacheService(redis)
    await cache_service.delete(f"user:{user_id}")
    
    return ApiResponse(
        success=True,
        message="用户删除成功",
        data={"deleted_user_id": user_id}
    )

# ============================================================================
# 11. MongoDB路由
# ============================================================================

@app.post("/mongo/users", response_model=ApiResponse)
async def create_mongo_user(
    user_data: MongoUser,
    mongo_db = Depends(get_mongo_db)
):
    """创建用户（MongoDB）"""
    service = MongoUserService(mongo_db)
    
    # 检查用户名是否已存在
    existing_user = await mongo_db.users.find_one({"username": user_data.username})
    if existing_user:
        raise HTTPException(status_code=400, detail="用户名已存在")
    
    user_id = await service.create_user(user_data.dict(exclude={"id"}))
    
    return ApiResponse(
        success=True,
        message="MongoDB用户创建成功",
        data={"user_id": user_id, "username": user_data.username}
    )

@app.get("/mongo/users/{user_id}", response_model=ApiResponse)
async def get_mongo_user(
    user_id: str,
    mongo_db = Depends(get_mongo_db)
):
    """获取用户（MongoDB）"""
    service = MongoUserService(mongo_db)
    user = await service.get_user_by_id(user_id)
    
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")
    
    return ApiResponse(
        success=True,
        message="获取MongoDB用户成功",
        data=user
    )

@app.get("/mongo/users", response_model=ApiResponse)
async def list_mongo_users(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100),
    search: Optional[str] = Query(None),
    mongo_db = Depends(get_mongo_db)
):
    """获取用户列表（MongoDB）"""
    service = MongoUserService(mongo_db)
    
    if search:
        users = await service.search_users(search)
    else:
        users = await service.get_users(skip, limit)
    
    return ApiResponse(
        success=True,
        message=f"获取MongoDB用户列表成功，共 {len(users)} 条记录",
        data={
            "users": users,
            "pagination": {
                "skip": skip,
                "limit": limit,
                "total": len(users)
            }
        }
    )

# ============================================================================
# 12. 缓存演示路由
# ============================================================================

@app.get("/cache/demo", response_model=ApiResponse)
async def cache_demo(
    redis: Redis = Depends(get_redis)
):
    """缓存演示"""
    cache_service = CacheService(redis)
    
    # 演示基本缓存操作
    await cache_service.set("demo_key", "Hello Redis!", expire=60)
    value = await cache_service.get("demo_key")
    
    # 演示JSON缓存
    demo_data = {
        "message": "这是一个JSON缓存示例",
        "timestamp": datetime.now().isoformat(),
        "data": [1, 2, 3, 4, 5]
    }
    await cache_service.set_json("demo_json", demo_data, expire=60)
    json_value = await cache_service.get_json("demo_json")
    
    # 演示计数器
    counter = await cache_service.increment("demo_counter")
    
    return ApiResponse(
        success=True,
        message="Redis缓存演示",
        data={
            "string_cache": value,
            "json_cache": json_value,
            "counter": counter,
            "operations": [
                "字符串缓存",
                "JSON缓存",
                "计数器",
                "过期时间设置"
            ]
        }
    )

@app.get("/cache/stats", response_model=ApiResponse)
async def cache_stats(
    redis: Redis = Depends(get_redis)
):
    """缓存统计信息"""
    try:
        info = await redis.info()
        
        return ApiResponse(
            success=True,
            message="Redis统计信息",
            data={
                "connected_clients": info.get("connected_clients", 0),
                "used_memory_human": info.get("used_memory_human", "0B"),
                "total_commands_processed": info.get("total_commands_processed", 0),
                "keyspace_hits": info.get("keyspace_hits", 0),
                "keyspace_misses": info.get("keyspace_misses", 0),
                "uptime_in_seconds": info.get("uptime_in_seconds", 0)
            }
        )
    except Exception as e:
        return ApiResponse(
            success=False,
            message=f"获取Redis统计信息失败: {str(e)}"
        )

# ============================================================================
# 13. 性能测试路由
# ============================================================================

@app.get("/performance/database-comparison", response_model=ApiResponse)
async def database_performance_comparison(
    db: AsyncSession = Depends(get_db),
    mongo_db = Depends(get_mongo_db),
    redis: Redis = Depends(get_redis)
):
    """数据库性能对比"""
    import time
    
    results = {}
    
    # SQLAlchemy性能测试
    start_time = time.time()
    repo = UserRepository(db)
    users = await repo.get_users(limit=10)
    sql_time = time.time() - start_time
    results["sqlalchemy"] = {
        "time_ms": round(sql_time * 1000, 2),
        "records": len(users),
        "type": "关系型数据库（SQLite）"
    }
    
    # MongoDB性能测试
    start_time = time.time()
    mongo_service = MongoUserService(mongo_db)
    mongo_users = await mongo_service.get_users(limit=10)
    mongo_time = time.time() - start_time
    results["mongodb"] = {
        "time_ms": round(mongo_time * 1000, 2),
        "records": len(mongo_users),
        "type": "文档数据库"
    }
    
    # Redis性能测试
    start_time = time.time()
    cache_service = CacheService(redis)
    for i in range(10):
        await cache_service.set(f"perf_test_{i}", f"value_{i}")
    redis_time = time.time() - start_time
    results["redis"] = {
        "time_ms": round(redis_time * 1000, 2),
        "operations": 10,
        "type": "内存数据库"
    }
    
    return ApiResponse(
        success=True,
        message="数据库性能对比完成",
        data={
            "results": results,
            "note": "性能结果仅供参考，实际性能取决于数据量、网络、硬件等因素"
        }
    )

# ============================================================================
# 14. 健康检查路由
# ============================================================================

@app.get("/health", response_model=ApiResponse)
async def health_check(
    db: AsyncSession = Depends(get_db),
    mongo_db = Depends(get_mongo_db),
    redis: Redis = Depends(get_redis)
):
    """健康检查"""
    health_status = {
        "database": False,
        "mongodb": False,
        "redis": False
    }
    
    # 检查SQLAlchemy数据库连接
    try:
        await db.execute(select(1))
        health_status["database"] = True
    except Exception as e:
        logging.error(f"Database health check failed: {e}")
    
    # 检查MongoDB连接
    try:
        await mongo_db.command("ping")
        health_status["mongodb"] = True
    except Exception as e:
        logging.error(f"MongoDB health check failed: {e}")
    
    # 检查Redis连接
    try:
        await redis.ping()
        health_status["redis"] = True
    except Exception as e:
        logging.error(f"Redis health check failed: {e}")
    
    all_healthy = all(health_status.values())
    
    return ApiResponse(
        success=all_healthy,
        message="健康检查完成" if all_healthy else "部分服务不可用",
        data={
            "status": health_status,
            "timestamp": datetime.now().isoformat(),
            "overall_status": "healthy" if all_healthy else "unhealthy"
        }
    )

# ============================================================================
# 15. 启动配置
# ============================================================================

if __name__ == "__main__":
    import uvicorn
    
    print("🚀 启动FastAPI数据库操作示例...")
    print("📚 API文档: http://localhost:8000/docs")
    print("🔍 交互式文档: http://localhost:8000/redoc")
    
    uvicorn.run(
        "05_database_comparison:app",
        host="0.0.0.0",
        port=8000,
        reload=True,
        log_level="info"
    )

# ============================================================================
# 16. FastAPI vs SpringBoot 数据库操作对比总结
# ============================================================================

"""
FastAPI vs SpringBoot 数据库操作详细对比:

1. ORM框架:
   - FastAPI: SQLAlchemy 2.0 (异步支持)
   - SpringBoot: JPA/Hibernate, MyBatis

2. 异步支持:
   - FastAPI: 原生异步，async/await
   - SpringBoot: R2DBC (响应式), CompletableFuture

3. 数据验证:
   - FastAPI: Pydantic模型，自动验证
   - SpringBoot: Bean Validation, @Valid注解

4. 依赖注入:
   - FastAPI: Depends()函数
   - SpringBoot: @Autowired, @Inject注解

5. 数据库连接:
   - FastAPI: 异步引擎，会话管理
   - SpringBoot: 连接池，事务管理

6. 缓存支持:
   - FastAPI: 手动集成Redis等
   - SpringBoot: @Cacheable注解，Spring Cache

7. 事务管理:
   - FastAPI: 手动管理事务
   - SpringBoot: @Transactional注解

8. 数据库迁移:
   - FastAPI: Alembic
   - SpringBoot: Flyway, Liquibase

9. 查询构建:
   - FastAPI: SQLAlchemy Core/ORM
   - SpringBoot: JPQL, Criteria API, QueryDSL

10. 性能特点:
    - FastAPI: 高并发，低延迟
    - SpringBoot: 成熟稳定，功能完整

11. 学习曲线:
    - FastAPI: 相对简单，Python生态
    - SpringBoot: 较复杂，Java企业级

12. 生态系统:
    - FastAPI: 轻量级，灵活组合
    - SpringBoot: 完整生态，开箱即用

13. 适用场景:
    - FastAPI: API服务，微服务，快速开发
    - SpringBoot: 企业应用，复杂业务逻辑

14. 开发效率:
    - FastAPI: 快速开发，自动文档
    - SpringBoot: 规范化开发，工具链完善

15. 总结:
    - FastAPI: 现代化、高性能的API框架
    - SpringBoot: 成熟、稳定的企业级框架
"""